package com.distributer.message.producer;

import java.util.concurrent.ArrayBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import com.distributer.message.model.Message;

/**
 *线程懒初始化，先将消息交给消息发送线程，消息发送线程判断该主题是否存在消息存储队列，
 *1：如果已经消息队列，将需要发送的消息存储到消息队列即可。
 *2：如果不存在消息队列，需要初始化一个消息队列，然后初始化一个线程用于处理该消息队列的消息。
 */
@Component
public class MessageRequestHander {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(MessageRequestHander.class);
	
	private static final int DEFAULT_QUEUE_SIZE = 2048; 
	
	public void messageRequestHander(Message message, KafkaTemplate<String, String> kafkaTemplate){
		try {
			MessageQueue messageQueue = MessageQueue.getInterface();
			//已经存在队列，直接将消息添加到队列即可
			if (messageQueue.exist(message.getNotifyTopic())) {
				ArrayBlockingQueue<Message> queue = messageQueue.getQueueByTopic(message.getNotifyTopic());
				queue.put(message);
				return;
			}
			synchronized (MessageRequestHander.class) {
				//再次判断是否存在
				if (messageQueue.exist(message.getNotifyTopic())) {
					ArrayBlockingQueue<Message> queue = messageQueue.getQueueByTopic(message.getNotifyTopic());
					queue.put(message);
					return;
				}
				//不存在队列，需要重新添加队列并且初始化一个线程，用于处理队列的任务
				ArrayBlockingQueue<Message> initQueue = new ArrayBlockingQueue<Message>(DEFAULT_QUEUE_SIZE);
				ProducerThread producerTask = new ProducerThread(initQueue,kafkaTemplate);
				final Thread producerThread = new Thread(producerTask, "thread-"+message.getNotifyTopic());
				producerThread.start();
				Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
					@Override
					public void run() {
						producerThread.interrupt();
						LOGGER.info("线程："+producerThread.getName()+"被中断！");
					}
				}));
				messageQueue.addTopicMapping(message.getNotifyTopic(), initQueue);
				initQueue.put(message);
			}
			return;
 		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}
